package com.weimin.mqtt.server.service;

import com.weimin.mqtt.server.handler.MqttChannelInboundHandler;
import com.weimin.mqtt.server.management.MqttSessionManager;
import com.weimin.mqtt.server.model.MqttServerConfig;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.mqtt.MqttDecoder;
import io.netty.handler.codec.mqtt.MqttEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

import java.io.File;

@Slf4j
public class MqttServerService {
    private NioEventLoopGroup bossGroup;
    private NioEventLoopGroup workGroup;
    // 客户端存储
    private MqttSessionManager mqttSessionManager;
    // 消息队列处理线程数
    private SslContext sslContext = null;
    private ChannelFuture channelFuture;

    public void start(MqttServerConfig config, MqttSessionManager sessionManager) {
        if (config == null) {
            return;
        }
        try {
            // 客户端管理数据
            mqttSessionManager = sessionManager;
            // 连接线程处理
            bossGroup = new NioEventLoopGroup(config.getBossThreadNumber());
            // 消息处理线程
            workGroup = new NioEventLoopGroup(config.getWorkThreadNumber());
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(bossGroup, workGroup);
            //设置通道模式为非阻塞Server Socket
            bootstrap.channel(NioServerSocketChannel.class);
            //SO_REUSEADDR允许在同一端口上启动同一服务器的多个实例，只要每个实例捆绑一个不同的本地IP地址即可
            bootstrap.option(ChannelOption.SO_REUSEADDR, false);
            //标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度
            bootstrap.option(ChannelOption.SO_BACKLOG, 1024);
            //禁用了Nagle算法，允许小包的发送
            bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
            //开启心跳保活机制，缺省时间很长，2小时。IdleStateHandler更合适。
            bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
            //非JVM管理的，由os管理的内存,减少数据传输的一次copy
            bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
            bootstrap.handler(new LoggingHandler(LogLevel.INFO));
            // 创建SSL上下文
            if (config.isSsl()) {
                sslContext = SslContextBuilder.forServer(new File(config.getServerCertFile()), new File(config.getKeyFile()))
                        .trustManager(new File(config.getCaCertFile())).build();
            }
            bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    ChannelPipeline channelPipeline = ch.pipeline();
                    if (sslContext != null) {
                        // 将SSL上下文添加到ChannelPipeline中
                        channelPipeline.addLast(sslContext.newHandler(ch.alloc()));
                    }
                    // 设置读写空闲超时时间，单位是秒.这里只考虑读，也就是channelRead() 方法超过 readerIdleTime 时间未被调用则会触发超时事件调用 userEventTrigger()。
                    channelPipeline.addLast(new IdleStateHandler(config.getCheckOfflineInterval(), 0, 0));
                    channelPipeline.addLast("encoder", MqttEncoder.INSTANCE);
                    channelPipeline.addLast("decoder", new MqttDecoder(config.getMaxBytesInMessage()));
                    channelPipeline.addLast(new MqttChannelInboundHandler(config, mqttSessionManager));
                }
            });
            channelFuture = bootstrap.bind(config.getPort()).sync();
//            f.channel().closeFuture().sync();
            Runtime.getRuntime().addShutdownHook(new Thread(this::close, "netty-close-thread"));
            log.info("MQTT 服务启动 端口 {}", config.getPort());
        } catch (Exception e) {
            log.error("MQTT 启动失败", e);
        }
    }

    /**
     * 手动关闭mqtt监听
     */
    public void close() {
        if (workGroup != null && bossGroup != null) {
            workGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
        if (channelFuture != null) {
            ReferenceCountUtil.release(channelFuture);
        }
    }
}
